package com.health.trans.job;

import cn.hutool.core.collection.CollUtil;
import com.health.trans.kit.BatchSendInfo;
import com.health.trans.pojo.JobConfig;
import com.health.trans.pojo.TableInfo;
import com.health.trans.pojo.TableRes;
import com.health.trans.pojo.TransParam;
import com.health.trans.processor.listener.ItemCountItemStream;
import com.health.trans.util.DataSourceUtil;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.transaction.PlatformTransactionManager;

import java.util.List;
import java.util.function.Consumer;
import java.util.stream.Collectors;

import static com.health.trans.service.BatchTransService.myTaskExecutor;

/**
 * @author nqlz
 */
public interface AbstractJob {
    org.slf4j.Logger log = org.slf4j.LoggerFactory.getLogger(AbstractJob.class);

    /**
     * 让消费
     * 获取相应处理方法
     *
     * @param param     参数
     * @param tableInfo 表信息
     * @return {@link T}
     */
    static <T> Consumer<JobConfig<T>> getSolveConsume(TransParam param, TableInfo<T> tableInfo) {
        return jobConfig -> {
            jobConfig.getJobInit().setIsParallel(param.getIsParallel());
            jobConfig.getJobInit().setParallelNum(param.getParallelNum());
            String readerName = param.getReader().getName();
            String writerName = param.getWriter().getName();
            String writerSource = param.getWriter().getSource();

            ItemWriter<T> writer = ItemCs.getWriter(writerName, tableInfo.getWriterConfig());
            jobConfig.initWriter(writer);

            ItemReader<T> reader = ItemCs.getReader(readerName, tableInfo.getReaderConfig());
            jobConfig.initReader(reader);

            jobConfig.initJobStep(s -> {
                PlatformTransactionManager transactionManager = DataSourceUtil.getTransactionManager(writerSource);
                return s.transactionManager(transactionManager);
            });

        };
    }

    /**
     * 处理数据
     *
     * @param param     参数
     * @param tableInfo 表信息
     * @return {@link TableRes}
     */
    static <T> TableRes solveData(TransParam param, TableInfo<T> tableInfo) {
        return new BatchSendInfo<>(param, tableInfo).sendAction();
    }

    /**
     * springBatch  处理默认实现
     * 需要多线程或其他方式实现 重写此方法即可
     *
     * @param param 参数
     * @return {@link List}<{@link TableRes}>
     */
    static <T> List<TableRes> execute(TransParam param) {
        List<TableInfo<T>> tables = param.getTables();
        log.info("需要处理表量：" + CollUtil.size(tables));
        return tables.stream().map(i -> {
            try {
                return solveData(param, i);
            } catch (Exception e) {
                log.error("异常：", e);
                return new TableRes().error(e.getMessage());
            }
        }).collect(Collectors.toList());
    }


}
